package com.mai.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mai.realtime.app.BaseAppV1;
import com.mai.realtime.bean.VideoPlayCountBean;
import com.mai.realtime.common.Constant;
import com.mai.realtime.function.DimAsyncFunction;
import com.mai.realtime.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsVideoPlayCountWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsVideoPlayCountWindow().init(
                4014,
                1,
                "Edu_14_DwsVideoPlayCountWindow",
                Constant.TOPIC_DWD_TRAFFIC_APPVIDEO
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //封装
        SingleOutputStreamOperator<VideoPlayCountBean> videoPlayPojo = VideoToPojo(stream);

        // 2:判断是否是一次独立的播放
        // 播放日志生成规则为：每过30s或关闭视频生成一条，从打开视频到关掉视频视为一次播放
        SingleOutputStreamOperator<VideoPlayCountBean> videoPlayCountBean = onevideoPlay(videoPlayPojo);


        // 3:判断独立访客(粒度天，使用状态保存时间)
        SingleOutputStreamOperator<VideoPlayCountBean> uniqueUserPlayCount = uniqueAggWindow(videoPlayCountBean);


        // 4:补充维度信息（ChapterId+ChapterName）
        SingleOutputStreamOperator<VideoPlayCountBean> DimStream = addDim(uniqueUserPlayCount);

        // 5:开窗聚合
        SingleOutputStreamOperator<VideoPlayCountBean> videoPlayCountAggWindow = VideoPlayAggWindow(DimStream);


        // 6:写出到Doris
        videoPlayCountAggWindow
                .map(bean-> JSON.toJSONString(bean))
                .print();
        //.addSink(FlinkSinkUtil.getDorisSink("gmall.dws_video_play_count_window"));
    }

    private SingleOutputStreamOperator<VideoPlayCountBean> VideoPlayAggWindow(SingleOutputStreamOperator<VideoPlayCountBean> dimStream) {
        return dimStream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<VideoPlayCountBean>() {
                            @Override
                            public VideoPlayCountBean reduce(VideoPlayCountBean bean1,
                                                             VideoPlayCountBean bean2) throws Exception {

                                bean1.setPlayCount(bean1.getPlayCount() + bean2.getPlayCount());
                                bean1.setPlayTimeCount(bean1.getPlayTimeCount() + bean2.getPlayTimeCount());
                                bean1.setPlayUserUniqueCount(bean1.getPlayUserUniqueCount() + bean2.getPlayUserUniqueCount());

                                return bean1;
                            }
                        },
                        new ProcessAllWindowFunction<VideoPlayCountBean, VideoPlayCountBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx,
                                                Iterable<VideoPlayCountBean> elements,
                                                Collector<VideoPlayCountBean> out) throws Exception {
                                VideoPlayCountBean bean = elements.iterator().next();
                                long start = ctx.window().getStart();
                                long end = ctx.window().getEnd();

                                bean.setStt(AtguiguUtil.toDateTime(start));
                                bean.setEdt(AtguiguUtil.toDateTime(end));
                                bean.setCurDate(AtguiguUtil.toDate(end));


                                out.collect(bean);
                            }
                        }
                );
    }

    private SingleOutputStreamOperator<VideoPlayCountBean> addDim(SingleOutputStreamOperator<VideoPlayCountBean> uniqueUserPlayCount) {
        SingleOutputStreamOperator<VideoPlayCountBean> dimStream = AsyncDataStream
                .unorderedWait(
                        uniqueUserPlayCount,
                        new DimAsyncFunction<VideoPlayCountBean>() {
                            @Override
                            public String getTable() {
                                return "dim_video_info";
                            }

                            @Override
                            public String getId(VideoPlayCountBean input) {
                                return input.getVideoId();
                            }

                            @Override
                            public void addDim(JSONObject dim, VideoPlayCountBean input) {
                                input.setChapterId(dim.getString("CHAPTER_ID"));
                            }
                        },
                        60,
                        TimeUnit.SECONDS
                );

        return AsyncDataStream
                .unorderedWait(
                        dimStream,
                        new DimAsyncFunction<VideoPlayCountBean>() {
                            @Override
                            public String getTable() {
                                return "DIM_CHAPTER_INFO";
                            }

                            @Override
                            public String getId(VideoPlayCountBean input) {
                                return input.getChapterId();
                            }

                            @Override
                            public void addDim(JSONObject dim, VideoPlayCountBean input) {

                                input.setChapterName(dim.getString("CHAPTER_NAME"));
                            }
                        },
                        60,
                        TimeUnit.SECONDS
                );
    }

    private SingleOutputStreamOperator<VideoPlayCountBean> uniqueAggWindow(SingleOutputStreamOperator<VideoPlayCountBean> videoPlayCountBean) {
        return videoPlayCountBean
                .keyBy(bean -> bean.getUid())
                .process(new KeyedProcessFunction<String, VideoPlayCountBean, VideoPlayCountBean>() {
                    private ValueState<String> todayState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        todayState = getRuntimeContext().getState(new ValueStateDescriptor<String>("", String.class));
                    }

                    @Override
                    public void processElement(VideoPlayCountBean bean,
                                               Context ctx,
                                               Collector<VideoPlayCountBean> out) throws Exception {
                        String today = AtguiguUtil.toDate(bean.getTs());
                        if (!today.equals(todayState.value())) {
                            bean.setPlayUserUniqueCount(1L);
                        }
                        out.collect(bean);

                    }
                });
    }

    private SingleOutputStreamOperator<VideoPlayCountBean> onevideoPlay(SingleOutputStreamOperator<VideoPlayCountBean> videoPlayPojo) {
        return videoPlayPojo
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VideoPlayCountBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getChapterId() + "_" + bean.getUid())
                .process(new KeyedProcessFunction<String, VideoPlayCountBean, VideoPlayCountBean>() {
                    private ValueState<VideoPlayCountBean> maxTimePalyState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 状态存储时间最大（最近时间）的一条
                        maxTimePalyState = getRuntimeContext().getState(new ValueStateDescriptor<VideoPlayCountBean>("MaxTimePalyState", VideoPlayCountBean.class));
                    }

                    @Override
                    public void processElement(VideoPlayCountBean bean,
                                               Context ctx,
                                               Collector<VideoPlayCountBean> out) throws Exception {
                        VideoPlayCountBean state = maxTimePalyState.value();

                        Long ts = bean.getTs();
                        // 如果maxTs==null时
                        if (state == null) {
                            maxTimePalyState.update(bean);
                            // 第一条通过播放时长判断：如果<30s则已经关闭,如果>30则还会有第二条
                            if (bean.getPlayTimeCount() < 30) {
                                bean.setPlayCount(1L);
                            }
                            // =30s可以注册定时器30再判断有没有下一条
                        } else {
                            // 状态不为空，可能是连续的播放，需要判断
                            if (ts - state.getTs() > 30) {
                                bean.setPlayCount(1L);
                            }
                        }
                        out.collect(bean);
                    }
                });
    }

    private SingleOutputStreamOperator<VideoPlayCountBean> VideoToPojo(DataStreamSource<String> stream) {
        return stream
                /*    // 如果维度数据有问题 videoId过滤掉>1002的
                    .filter(json->{
                        JSONObject obj = JSON.parseObject(json);
                        return obj.getJSONObject("appVideo").getLong("video_id")<1003;
                    })
    */
                .map(json -> {

                    JSONObject obj = JSON.parseObject(json);

                    JSONObject common = obj.getJSONObject("common");
                    JSONObject appVideo = obj.getJSONObject("appVideo");

                    Long ts = obj.getLong("ts");

                    // 视频播放时长
                    Long play_sec = appVideo.getLong("play_sec");

                    String video_id = appVideo.getString("video_id");

                    String uid = common.getString("uid");


                    return new VideoPlayCountBean(
                            uid, video_id, "", "",
                            "", "", "",
                            0L,
                            play_sec,
                            0L, ts
                    );
                });
    }
}
